package com.adam.example.aggregation.pipeline;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.aggregations.*;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.util.DateTime;
import com.adam.example.ElasticsearchClientHolder;
import com.adam.example.util.ReflectUtil;

import java.io.IOException;
import java.time.format.DateTimeFormatter;
import java.util.List;

public class CumulativeCardinalityAggregationExample {

    public static void main(String[] args) throws IOException {
        ElasticsearchClient elasticsearchClient = ElasticsearchClientHolder.getINSTANCE().initializeAndGetClient();

        SearchResponse<Void> searchResponse = elasticsearchClient.search(builder -> builder
                        .index("test_user_hits")
                        .size(0)
                        .aggregations("users_per_day", aggregationBuilder -> aggregationBuilder
                                .dateHistogram(dateHistogramAggregationBuilder -> dateHistogramAggregationBuilder
                                        .field("date")
                                        .calendarInterval(CalendarInterval.Day)
                                )
                                .aggregations("distinct_users", aggregationBuilder1 -> aggregationBuilder1
                                        .cardinality(cardinalityAggregationBuilder -> cardinalityAggregationBuilder
                                                .field("user_id")
                                        )
                                )
                                .aggregations("cumulative_distinct_users", aggregationBuilder2 -> aggregationBuilder2
                                        .cumulativeCardinality(cumulativeCardinalityAggregationBuilder -> cumulativeCardinalityAggregationBuilder
                                                .bucketsPath(bucketsPathBuilder -> bucketsPathBuilder
                                                        .single("distinct_users")
                                                )
                                        )
                                )
                        )
                , Void.class);
        Aggregate usersPerDayAggregate = searchResponse.aggregations().get("users_per_day");
        DateHistogramAggregate dateHistogramAggregate = usersPerDayAggregate.dateHistogram();
        Buckets<DateHistogramBucket> buckets = dateHistogramAggregate.buckets();
        List<DateHistogramBucket> bucketList = buckets.array();
        for(DateHistogramBucket bucket: bucketList) {
            DateTime dateTime = bucket.key();
            String formatted = dateTime.toZonedDateTime().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
            System.out.println(formatted + ":" + bucket.docCount());
            Aggregate distinctUsersAggregate = bucket.aggregations().get("distinct_users");
            CardinalityAggregate cardinalityAggregate = distinctUsersAggregate.cardinality();
            System.out.println("daily distinct users:" + cardinalityAggregate.value());
            Aggregate cumulativeDistinctUsersAggregate = bucket.aggregations().get("cumulative_distinct_users");
            try {
                CumulativeCardinalityAggregate cumulativeCardinalityAggregate = ReflectUtil.getPrivateField(
                        cumulativeDistinctUsersAggregate, "_value");
                System.out.println("cumulative distinct users:" + cumulativeCardinalityAggregate.value());
            } catch (NoSuchFieldException | IllegalAccessException e) {
                System.err.println("反射获取field失败");
                e.printStackTrace();
            }
        }

        ElasticsearchClientHolder.getINSTANCE().close();
    }

}
